package reactor;


import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.List;
import java.util.stream.Collectors;

public class ReactorApp {
//    终结操作
    public static void op2(String[] args) {
        Flux<Integer> flux1 = Flux.range(1, 10);
        Mono<Long> count = flux1.count();
        Mono<Integer> reduce = flux1.reduce(0, (a, b) -> a + b);
        List<Integer> list1 = flux1.collectList().block();
        //求平均
        Double avg1 = flux1.collect(Collectors.averagingInt(item->item.intValue())).block();
        System.out.println(avg1);
    }
//    非终结操作
    public static void op1(String[] args) {
        Flux<Integer> flux1 = Flux.range(1, 10);
        flux1
                .filter(item->item%2==0)
                .map(item->item*10)
                .sort((a,b)->b-a)
                .flatMap(item-> Flux.just(item,item/10))
                .subscribe(System.out::println);
    }
//    合并操作
    public static void zipTest(String[] args) throws Exception{
        Flux<Integer> flux1 = Flux.range(1, 1000);
        Flux<Object> flux2 = Flux.create(fluxSink -> {
            while (true) {
                fluxSink.next(System.currentTimeMillis());
            }
        }).sample(Duration.ofMillis(100));
        flux1.zipWith(flux2,(a,b)->a+":\t"+b).subscribe(System.out::println);
    }
}
